package com.lvmama.rhino.analyze.push

import com.lvmama.rhino.common.entity.JobConfig
import com.lvmama.rhino.common.utils.spark.SparkApplication
import org.apache.commons.codec.digest.DigestUtils
import org.apache.spark.sql.SQLContext

/**
  * @Title: InvalidDeviceInit
  * @Package com.lvmama.rhino.analyze.push
  * @Description: 初始化，用于初始化push_mi_regId、push_deviceInfo表中的最后一次登录时间和push_locationInfo,只执行一次
  * @author 吴川通
  * @date 2016/11/15 12:28
  * @version V1.0
  */
class InvalidDeviceInit(config: JobConfig) extends SparkApplication {
  override var appName: String = "InvalidDeviceInit"
  override var sparkConfig: Map[String, String] = config.spark
  val TBL_PUSH_DEVICEINFO: String = "push_deviceInfo"
  val TBL_PUSH_LOCATIONINFO: String = "push_locationInfo"
  val TBL_PUSH_MI_REGID: String = "push_mi_regId"
  val FAM_BASEINFO: String = "baseInfo"

  def execute(): Unit = {
    sparkConfig += ("spark.app.name" -> appName)

    withSparkContext { sc =>

      val sqlContext = new SQLContext(sc)
      import com.lvmama.rhino.common.utils.hbase._
      import sqlContext.implicits._

      val miRegIdRdd = sc.hbaseTable[(String, Option[String], Option[String], Option[String], Option[String], Option[String])](TBL_PUSH_DEVICEINFO)
        .select("objectId", "mipushRegId", "mProvinceId", "platform", "userId").inColumnFamily(FAM_BASEINFO)

      //1.初始化push_deviceInfo第一次登录时间和最后一次登录时间
      val deviceInfoRdd = sc.hbaseTable[(String, Option[String], Option[String], Option[String])](TBL_PUSH_DEVICEINFO).select("objectId", "createTime", "lastTime").inColumnFamily(FAM_BASEINFO)
      deviceInfoRdd.filter(f => f._3.isEmpty || f._3.get == "").map(p => (p._1, "2016-11-01 00:00:00", "2016-11-01 00:00:00"))
        .toHBaseTable(TBL_PUSH_DEVICEINFO).toColumns("createTime", "lastTime").inColumnFamily(FAM_BASEINFO).save()

      //2.初始化push_mi_regId表
      //使用md5对小米id进行哈希
      miRegIdRdd.filter(f => f._3.nonEmpty && f._5.nonEmpty).map(f => (DigestUtils.md5Hex(f._3.get), f._4, f._2, f._5, f._6, f._3))
        .toHBaseTable(TBL_PUSH_MI_REGID).toColumns("mProvinceId", "objectId", "platform", "userId", "mipushRegId").inColumnFamily(FAM_BASEINFO).save()

      //3.初始化push_locationInfo表中的平台和有效性
      val locationInfoRdd = sc.hbaseTable[(String, Option[String])](TBL_PUSH_LOCATIONINFO)
        .select("objectId").inColumnFamily(FAM_BASEINFO).map(m => (m._1, m._2)).toDF("lKey", "objectId")

      val deviceInfo4LocationRdd = sc.hbaseTable[(String, Option[String], Option[String], Option[String], Option[String])](TBL_PUSH_DEVICEINFO)
        .select("objectId", "platform", "mipushRegId", "gCID").inColumnFamily(FAM_BASEINFO)
        .map(m => (m._1, m._2, m._3, if (m._4.isEmpty || m._4.get == "") "0" else "1", if (m._5.isEmpty || m._5.get == "") "0" else "1")).toDF("dKey", "objectId", "platform", "miValid", "gtValid")

      locationInfoRdd.join(deviceInfo4LocationRdd, Seq("objectId")).select("lKey", "platform", "miValid", "gtValid")
        .map(r => (r.get(0).toString, r.get(1).toString, r.get(2).toString, r.get(3).toString))
        .rdd
        .toHBaseTable(TBL_PUSH_LOCATIONINFO).toColumns("platform", "miValid", "gtValid").inColumnFamily(FAM_BASEINFO).save()


    }
  }

}

object InvalidDeviceInit {
  val config = JobConfig()

  def apply(): Unit = new InvalidDeviceInit(config).execute()

}

